/* **
 * Juan C. Caicedo
 * Universidad Nacional de Colombia
 * May, 2009
 * */

package m3f.computing;

import java.util.Arrays;
import java.util.concurrent.BrokenBarrierException;
import java.util.concurrent.CyclicBarrier;
import java.util.concurrent.Executor;
import java.util.concurrent.Executors;
import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.ThreadPoolExecutor;

public abstract class ConcurrentTemplate implements Runnable{
	
	/** ***********************
     * SUBTASK MANAGER CLASS 
     * ************************/
	// -----------------------------------------------
	// The following code does not need to be modified
	// -----------------------------------------------
	private class SubtaskManager{
		//static final int DOING_NOTHING = 0;
        //static final int COMPUTING = 1;
                
        /** **********************************
         * CLASS WORKER (DISTRIBUTABLE WORK)
         * ***********************************/
     
        private class Worker implements Runnable{
    		int start, num;
    		
    		public Worker(int s, int n, int i){
    			start = s;
    			num = n;
    		}

    		private void compute(){
    			int limit = start+num;
    			for(int i = start; i < limit; i++){
    				singleOperation(i);
    			}
    		}
    		
    		@Override
    		public void run() {
                try {
                	compute();
                } finally {
                    if (mBarrier != null) {
                        try {
                            mBarrier.await();
                        } catch (InterruptedException ex) {
                        } catch (BrokenBarrierException ex) {
                        }
                    }
                }
    		}
    	}
        private boolean mWorking;
        private Executor mExecutor;
        private CyclicBarrier mBarrier;
        
        private Worker[] mWorkers;
        
        SubtaskManager(int numThreads) {
        	mWorkers = new Worker[numThreads];
        	int[] regsPerWorker = new int[numThreads];
        	Arrays.fill(regsPerWorker, problemSize/numThreads);
            int leftOvers = problemSize - numThreads * regsPerWorker[0];
            for (int i = 0; i < leftOvers; i++) {
                regsPerWorker[i]++;
            }
            int startReg = 0;
            for (int i = 0; i < numThreads; i++) {
                mWorkers[i] = new Worker(startReg, regsPerWorker[i], i);
                startReg += regsPerWorker[i];
            }
            if (numThreads == 1) {
                mExecutor = new Executor() {
                    public void execute(Runnable runnable) {
                        if (!Thread.interrupted()) {
                            runnable.run();
                        } else {
                            throw new RejectedExecutionException();
                        }
                    }
                };
                
            } else {
                mBarrier = new CyclicBarrier(numThreads, new Runnable() {
                    public void run() {
                        workersDone();
                    }
                });
                mExecutor = Executors.newFixedThreadPool(numThreads);
            }
        }
        
        boolean compute() {
            boolean ok = false;
            mWorking = true;
            try {
                if (mBarrier != null) {
                    mBarrier.reset();
                }
                for (int i = 0; i < mWorkers.length; i++) {
                    mExecutor.execute(mWorkers[i]);
                }
                if (mBarrier != null) {
                    waitOnWorkers();
                    ok = !mBarrier.isBroken();
                } else {
                    ok = true;
                }
            } catch (RejectedExecutionException ree) {
                ree.printStackTrace();
            } finally {
                mWorking = false;
            }
            return ok;
        }
        
        void shutdown() {
            if (mExecutor instanceof ThreadPoolExecutor) {
                ((ThreadPoolExecutor) mExecutor).shutdownNow();
            }
        }
        
        private synchronized void waitOnWorkers() {
            while (mWorking) {
                try {
                    wait();
                } catch (InterruptedException ie) {
                    break;
                }
            }
        }
        
        private synchronized void workersDone() {
            mWorking = false;
            notifyAll();
        }
	}
	private int problemSize;
	private SubtaskManager mSubtaskManager;
	
	private int workers;
	
	public ConcurrentTemplate(int problemSize){
		this.problemSize = problemSize;
		this.workers = 0;
	}
	
    private void cleanup() {
        if (mSubtaskManager != null) {
            mSubtaskManager.shutdown();
            mSubtaskManager = null;
        }
    }
    
    public int getProblemSize(){
    	return problemSize;
    }
    
    public void run(){
    	int cpus = workers();
    	mSubtaskManager = new SubtaskManager( cpus );
    	try{
    		mSubtaskManager.compute();
	    } catch (Throwable t) {
	        t.printStackTrace();
	    } finally {
	        cleanup();
	    }
    }
    
    public abstract void singleOperation(int reg);
	
    public int workers(){
    	if(workers == 0){
    		workers = Runtime.getRuntime().availableProcessors();
    	}
    	return workers;
    }
}
